import calendar
import sys
import time
import pykafka
from pykafka.common import OffsetType
from pykafka.protocol import PartitionOffsetCommitRequest
from datetime import datetime

def fetch_offsets(client, topic, offset):
    """Fetch raw offset data from a topic """
    if offset.lower() == 'earliest':
        return topic.earliest_available_offsets()
    elif offset.lower() == 'latest':
        return topic.latest_available_offsets()
    else:
        offset = datetime.strptime(offset, "%Y-%m-%dT%H:%M:%S")
        offset = int(calendar.timegm(offset.utctimetuple()) * 1000)
        return topic.fetch_offset_limits(offset)


def fetch_consumer_lag(client, topic, consumer_group):
    """Get raw lag data for a topic/consumer group """
    latest_offsets = fetch_offsets(client, topic, 'latest')
    consumer = topic.get_simple_consumer(consumer_group=consumer_group,
                                         auto_start=False)
    current_offsets = consumer.fetch_offsets()
    return {p_id: (latest_offsets[p_id].offset[0], res.offset) for p_id, res in current_offsets}


def output_topic_lag(client, topic):
    """topic is string """
    tmp_topic = client.topics[topic]
    return tmp_topic

client = pykafka.KafkaClient(hosts="")
topic_str = b"devicesnumber"
topic = output_topic_lag(client, topic_str)
group_zhu = b'ceshi'
mm = fetch_consumer_lag(client, topic, group_zhu)[0]

#logsize, offset, lag
print(mm[0], mm[1], mm[0] - mm[1])